package com.weipt.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class HelloWorkerPoll {

    public static final String QUEUE_NAME = "hello-worker-poll—queue";

    public static void main(String[] args) throws Exception {
        consume1();
        consume2();
        for (int i = 0; i < 20; i++) {
            publish(String.valueOf(i));
            //Thread.sleep(1000);
        }
    }

    public static void publish(String msg) throws IOException, TimeoutException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel(); //创建一个新的管道
        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        channel.basicPublish("", QUEUE_NAME,  MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
        //关闭连接
        channel.close();
        mqConn.close();
    }

    public static void consume1() throws IOException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        //自动应答当autoack是true的时候，代表谁消费，谁自动回执，属于轮询分发
        // 若干消费者，即使性能不一，但也是逐个分发，适用于消费者集群稳定，配置一样的情况
        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumeA："+new String(body));
            }
        });
    }

    public static void consume2() throws IOException {
        Connection mqConn = ConnectionUtils.getMqConn();
        Channel channel = mqConn.createChannel();
        channel.queueDeclare(QUEUE_NAME, false,false,false,null);
        //Direct模式下是没有routingkey的，所以下面的第二个参数，实际是queueName
        channel.basicConsume(QUEUE_NAME, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumeB："+new String(body));
            }
        });
    }
}
